package tinynio;

import java.io.Closeable;
import java.io.IOException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Select implements Executor, Runnable, Closeable {
	private ExecutorService service = Executors.newSingleThreadExecutor();
	private Queue<Runnable> taskQueue = new ConcurrentLinkedQueue<Runnable>();
	private volatile boolean start = false;

	private Selector selector;
	private SelectGroup parent;

	public Select(SelectGroup parent) throws IOException {
		this.parent = parent;
		this.selector = Selector.open();
	}

	@Override
	public void execute(Runnable command) {
		if (!start)
			Select.this.run();
		taskQueue.offer(command);
	}

	@Override
	public void run() {
		service.submit(new Runnable() {
			@Override
			public void run() {
				startThread();
			}
		});
	}

	private void startThread() {
		assert !start;

		System.out.println("start thread");

		start = true;
		while (start) {
			try {
				if (!taskQueue.isEmpty()) {
					selector.selectNow();
				} else
					selector.select(100);

				if (!start)
					break;

				Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
				while (iter.hasNext()) {
					SelectionKey key = iter.next();
					iter.remove();
					if (!key.isValid()) {
						SocketCommunicator sc = (SocketCommunicator) key.attachment();
						sc.close();
						continue;
					}

					if (key.isAcceptable()) {
						doAccept(key);
					}
					if (key.isConnectable()) {
						doConnect(key);
					}
					if (key.isReadable()) {
						System.out.println("do read");
						doRead(key);
					}
					if (key.isWritable()) {
						System.out.println("do isWritable");
						doWrite(key);
					}
				}

				runAllTask();
			} catch (CancelledKeyException c) {
				System.out.println("ignore CancelledKeyException");
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	private void doWrite(SelectionKey key) {
		SocketCommunicator sc = (SocketCommunicator) key.attachment();
		sc.flush();
	}

	private void doRead(SelectionKey key) throws IOException {
		SocketCommunicator sc = (SocketCommunicator) key.attachment();
		sc.read();
	}

	private void doConnect(SelectionKey key) throws IOException {
		SocketChannel channel = (SocketChannel) key.channel();
		if (channel.finishConnect()) {
			key.interestOps(key.interestOps() & ~SelectionKey.OP_CONNECT);

			if (isActive()) {
				SocketCommunicator sc = (SocketCommunicator) key.attachment();
				sc.handler().onActive(sc);
			}
		}

	}

	private void doAccept(SelectionKey key) throws IOException {
		System.out.println("accept");
		ServerSocketChannel ch = (ServerSocketChannel) key.channel();
		SocketChannel sc = ch.accept();
		SocketCommunicator communicator = new SocketCommunicator(parent.next(), sc);
		((ServerSocketCommunicator) key.attachment()).config(communicator);
		communicator.register();

		if (communicator.isActive())
			communicator.handler().onActive(communicator);
	}

	private void runAllTask() {
		while (true) {
			if (taskQueue.isEmpty())
				return;

			Runnable task = taskQueue.poll();
			if (task != null)
				task.run();
		}

	}

	@Override
	public void close() {
		try {
			start = false;
			service.shutdown();
			taskQueue.clear();

			selector.wakeup();
			selector.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public Executor service() {
		return service;
	}

	public boolean isActive() {
		return start;
	}

	public Selector getSelector() {
		return selector;
	}
}
